package com.navinfo.tripanalysis.offline.accumulator;

import com.navinfo.tripanalysis.offline.util.AccumulatorUtils;
import org.apache.spark.util.AccumulatorV2;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 0f37异常数据累加器，统计各项异常指标
 * @author hmc
 */

public class Load0F37DataAccumulator extends AccumulatorV2<String,String> {

    private static final Logger logger = LoggerFactory.getLogger(Load0F37DataAccumulator.class);

    //设置默认累加值
    private static final int ACCUMULATOR_VALUE = 1;

    //设置各统计指标原始状态
    /**
     * 频次：车/天
     * count：总车辆数
     * repeated_count: 存在gpstime重复点车辆数
     * speed_10_100_count：速度超过150km/h的点数量在10-100之间的车辆数
     * speed_10_100_count：速度超过150km/h的点数量在100-300之间的车辆数
     * speed_10_100_count：速度超过150km/h的点数量在300以上的车辆数
     * rotation_10_100_count：转速超过3000r/min的点数量在10-100之间的车辆数
     * rotation_100_300_count：转速超过3000r/min的点数量在100-300之间的车辆数
     * rotation_300_count：转速超过3000r/min的点数量在300以上的车辆数
     */
    private String zero = DCConstants.ACC_0F37_COUNT+"=0|"
                            + DCConstants.ACC_0F37_REPEATED_0_10_Ratio+"=0|"
                            + DCConstants.ACC_0F37_REPEATED_10_50_Ratio+"=0|"
                            + DCConstants.ACC_0F37_REPEATED_50_Ratio+"=0|"
                            + DCConstants.ACC_0F37_SPEED_10_100_COUNT+"=0|"
                            + DCConstants.ACC_0F37_SPEED_100_300_COUNT+"=0|"
                            + DCConstants.ACC_0F37_SPEED_300_COUNT+"=0|"
                            + DCConstants.ACC_0F37_ROTATION_10_100_COUNT+"=0|"
                            + DCConstants.ACC_0F37_ROTATION_100_300_COUNT+"=0|"
                            + DCConstants.ACC_0F37_ROTATION_300_COUNT+"=0"
                            ;

    //初始化结果
    private String data = zero;

    @Override
    public boolean isZero() {
        return data.equalsIgnoreCase(zero);
    }

    @Override
    public AccumulatorV2<String, String> copy() {
        return new Load0F37DataAccumulator();
    }

    @Override
    public void reset() {
        data = zero;
    }

    @Override
    public void add(String v) {
        data = AccumulatorUtils.mergeData(data,"\\|",v,ACCUMULATOR_VALUE);
    }

    @Override
    public void merge(AccumulatorV2<String, String> other) {
        data = AccumulatorUtils.mergeAccumulator(data,"\\|",other);
    }

    @Override
    public String value() {
        return data;
    }



}
